package com.liyu.rabbit.roll;

import com.liyu.rabbit.Producer;
import com.liyu.rabbit.factory.RabbitFactory;
import com.rabbitmq.client.Channel;

import java.io.IOException;
import java.util.concurrent.*;

/**
 * @author 李誉
 * @date 2022-05-16 20:07
 * @description 公平调度
 */
public class FairConsumer {

    public static void main(String[] args) {

        ExecutorService executorService = new ThreadPoolExecutor(
                2,
                3,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(2),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );


        for (int i=0; i < 2; i++) {
            int index = i;
            executorService.execute( () -> {
                try {
                    // 创建通道
                    Channel newChannel = RabbitFactory.createNewChannel();

                    newChannel.basicQos(1);
                    // 消费逻辑
                    newChannel.basicConsume(Producer.QUEUE_NAME, false, (consumerTag, message) -> {
                        System.out.println("线程 -"+ index + "- 消费一条消息：" + new String(message.getBody()));
                        if (index % 2 == 0) {
                            try {
                                Thread.sleep(5000);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                        // 手动应答，第二个参数控制是否批量应答
                        newChannel.basicAck(message.getEnvelope().getDeliveryTag(), false);
                        System.out.println("线程 -"+ index + "- 消费消息成功:"+ new String(message.getBody()));
                    }, (consumerTag) -> {
                        System.out.println("线程 -"+ index + "- 消费一条消息失败："+consumerTag);
                    });
                } catch (IOException | TimeoutException e) {
                    e.printStackTrace();
                }

            });
        }
    }
}
